package com.atguigu.flink.watermark;

import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/23
 *
 *   和基于事件时间的窗口结合
 *
 *      窗口是窗口(window):  数据是否落入某个窗口，参考数据的eventtime是否在窗口的时间范围内
 *                              [start,end)
 *                              2022-11-23 10:01:00 -   2022-11-23 10:02:00
 *      水印是水印(waterwindow)
 *          没有关系，只有一丝间接联系。 水印更新算子的时钟，基于时间的窗口，时间到了，会触发窗口的运算。
 *                      水印 --------->更新时钟 ---------->到点触发运算。
 */
public class Demo2_EventTimeWindow
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.getConfig().setAutoWatermarkInterval(2000);

        env.setParallelism(1);

        //水印策略
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()   //目前的场景数据都是有序的
                                                      //如何产生水印
                                                      .withTimestampAssigner((e, r) -> e.getTs());

        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           .assignTimestampsAndWatermarks(watermarkStrategy)
           //第一个窗口:  [0,4999] 或 [0,5000)
           //第二个窗口:  [5000,9999]
           .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
           .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>()
           {
               @Override
               public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                   TimeWindow window = context.window();

                   out.collect(window + "数据:"+MyUtil.parseList(elements).toString());
               }
           })
           .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
